package com.atguigu.day05;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * 使用union合流后  使用coFlatMapFunction合流
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> queryStream = env.socketTextStream("192.168.1.102", 9999);
        env.addSource(new com.atguigu.day02.Example3.MySource())
                .keyBy(r->r.name)
                .connect(
                        queryStream.broadcast()
                )
                .flatMap(new CoFlatMapFunction<com.atguigu.day02.Example3.Event, String, com.atguigu.day02.Example3.Event>() {
                    private String query = "";
                    @Override
                    public void flatMap1(com.atguigu.day02.Example3.Event value, Collector<com.atguigu.day02.Example3.Event> out) throws Exception {
                        if (query.equals(value.url)){
                            out.collect(value);
                        }
                    }

                    @Override
                    public void flatMap2(String value, Collector<com.atguigu.day02.Example3.Event> out) throws Exception {
                        query=value;
                    }
                })
                .print();

        env.execute();
    }
}
